package com.gjy.pulsar.resolve.producer;

import com.gjy.pulsar.resolve.config.PulsarClientConfig;
import com.gjy.pulsar.resolve.util.IdGenerator;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * 幂等生产者（开启 Topic 级去重、持久化发送、自动重试）。
 * 适用于“需要保证不丢失、避免因网络重试产生的重复写入”场景。
 *
 * @author gjy
 * @version 1.0
 * @since 2025-10-13 15:40:11
 */
public class IdempotentProducer implements AutoCloseable {
    private static final Logger logger = LoggerFactory.getLogger(IdempotentProducer.class);

    private final Producer<byte[]> producer;

    /**
     * @param topic 已经在 Broker 中创建且开启 duplicateDetection 的持久化主题
     * @throws PulsarClientException 初始化异常
     */
    public IdempotentProducer(String topic) throws PulsarClientException {
        PulsarClient client = PulsarClientConfig.newClient();

        this.producer = client.newProducer()
                .topic(topic)
                .enableBatching(true)                     // 批量提高吞吐
                .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
                .blockIfQueueFull(true)                    // 队列满时阻塞，防止丢失
                .sendTimeout(30, TimeUnit.SECONDS)        // 超时后自动重试
                .maxPendingMessages(1000)
                // 关键：为每条消息设置唯一的 producerName + sequenceId，实现去重
                .producerName(IdGenerator.uuid())          // 全局唯一的 Producer 名称
                .enableChunking(true)                      // 大消息分片
                .create();
        logger.info("IdempotentProducer 已创建，topic={}", topic);
    }

    /**
     * 同步发送（业务侧需要确认写入成功时使用）。
     *
     * @param key   可选的消息键（用来做 Key_Shared 时的路由）
     * @param value 消息体（已序列化为 byte[]）
     */
    public void sendSync(String key, byte[] value) throws PulsarClientException {
        MessageId msgId = producer.newMessage()
                .key(key)
                .value(value)
                // 为每条消息显式设置唯一 sequenceId（可选；如果未设置，则使用内部自增）
                .property("seqId", IdGenerator.nextLongString())
                .send();   // 阻塞直到 broker 确认写入
        logger.info("消息写入成功，msgId={}, key={}", msgId, key);
    }

    /**
     * 异步发送（业务不需要立即阻塞）。
     *
     * @param key   消息键
     * @param value 消息体
     */
    public void sendAsync(String key, byte[] value) {
        producer.newMessage()
                .key(key)
                .value(value)
                .property("seqId", IdGenerator.nextLongString())
                .sendAsync()
                .whenComplete((msgId, ex) -> {
                    if (ex != null) {
                        logger.error("异步发送失败，key={}", key, ex);
                    } else {
                        logger.debug("异步发送成功，msgId={}, key={}", msgId, key);
                    }
                });
    }

    @Override
    public void close() throws PulsarClientException {
        try {
            producer.flush();
        } finally {
            producer.close();
            logger.info("IdempotentProducer 已关闭");
        }
    }

}
